package com.shujia.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo4SourceFunction {
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    /**
      * 读取自定义的数据
      *
      */

    val myDS: DataStream[Int] = env.addSource(new MySource)

    myDS.print()

    env.execute()

  }
}

/**
  * 自定义source,实现SourceFunction接口
  *
  */
class MySource extends SourceFunction[Int] {
  /**
    * run ： 用于生成数据的方法，只执行一次
    * ctx：用于将数据发送到下游的对象
    */
  override def run(ctx: SourceFunction.SourceContext[Int]): Unit = {

    var i = 0
    while (true) {

      i += 1
      //将数据发送到下游
      ctx.collect(i)

      Thread.sleep(100)
    }

  }


  //任务被取消的时候执行，一般用于回收资源
  override def cancel(): Unit = {

  }
}
